#include "Dispatcher.h"
#include "lib/json.hpp"
#include "module/biz.h"
using json = nlohmann::json;
/*
#include <iostream>
using std::endl;
using std::cout;*/

void Dispatcher::dispatcher_cycle()
{
    epoll_event event;
    connection_t* connections = nullptr;
    auto dispatcherLogger = utils.logger.setBasicLogger("Dispatcher");
    while(true){
        if(!utils.queue.Empty()){
            event = utils.queue.Read();
            if(event.data.ptr == nullptr){
                //log: unregistered connection received.
                dispatcherLogger->warn("收到数据域无效的连接");        
                continue;
            }
            connections = (connection_t*) event.data.ptr;

            if(connections->type == UNUSED || connections->type == TOBEUSED ){ //规定连接在被分配后必须指定其类型
                //log: UNUSED/TOBEUSED connection received
                dispatcherLogger->warn("接收到类型为UNUSED或者TOBEUSED的连接");
                continue;
            }
            if(connections->fd == utils.serverSocket){ //如果是侦听socket，重新获取一次事件
                continue;
            }
            if(event.events & EPOLLHUP){
                std::cout << "断开连接" <<std::endl;
                closeConnection(connections, utils);
            }
            else if(event.events & EPOLLIN){ //读事件
                if(connections -> type == LOCAL_CONNECTION){ //用户连接
                    if(connections->reader.get() == nullptr){
                        //std::cout << "接收到用户连接"<<std::endl;
                        X6* temp = new X6(); //初始化协议（这里小心内存泄漏！！！！）
                        connections->reader.reset(temp);
                    }
                    //workerPool.enqueue(&Dispatcher::localReadEventHandler,this, connections);          
                }
                else if(connections->type == PEER_CONNECTION)
                {
                   // std::cout << "接收到对端连接"<<std::endl;
                    if(connections->peerReader.get() == nullptr){
                        X7* temp = new X7();
                        connections->peerReader.reset(temp);
                    }
                }
                readEventHandler(connections);
            }
            else if(event.events & EPOLLOUT){ //写事件
                if(connections->writer.get() == nullptr){
                    //从epoll中删除写事件
                    connections->ev ^= EPOLLOUT; 
                    epoll_event ev = utils.IOMultiplexer.setEpollEvent(connections->fd, connections->ev, (void*)connections);
                    utils.IOMultiplexer.epollAddEvent(ev, connections->fd);
                    continue;
                }else{
                    writeEventHandler(connections);
                }
            }

        }else{
            //队列为空时，说明网络空闲，睡眠2s，交出处理器使用权，避免空转
            std::unique_lock<std::mutex> lck(utils.signal_lock);
            utils.conn.wait_for(lck, std::chrono::seconds(2));       
        }
    }
    
}

void Dispatcher::run()
{
    std::thread DispathcerThread(&Dispatcher::dispatcher_cycle, this);
    DispathcerThread.detach();
}

void Dispatcher::readEventHandler(connection_t* conn)
{
    auto logger = utils.logger.setBasicLogger("Dispatcher");
    Socket clientSocket(conn->fd, CLIENT_SOCKET); 
    int ret;
    char buffer[BUFFER_SIZE];
    
    // void* current = buffer;
    /*int bytes_read = 0;
    int bytes_left = BUFFER_SIZE;*/

    /**
     * 理论上讲这里应该使用非阻塞读，准备一个足够大的buffer，将数据全部读取后再呈递给业务层
     * 但是我们不知道用户数据有多长，我们的buffer是否足够大
     * 因此这里采用的边读取数据边做协议解析的方法
     * 读到不再可读为止
     * **/
    while(true){
        ret = clientSocket.Recv(buffer, BUFFER_SIZE);
        if(ret == 0){
            //有读事件但是没有数据就绪，说明用户已经断开连接
            //closeConnection(conn, utils); //关闭连接
            return;
        }else if(ret < 0){
            //发生错误
            if(clientSocket.getErrorno() == EAGAIN) break;
            else{
                logger->error("recv失败：{}", MyLogger::getSysError().c_str()); 
                closeConnection(conn,utils);
                return;
            }
        }else{
            int val = 0;
            if(conn->type == LOCAL_CONNECTION){
                val = localConnHandler(conn, buffer, ret);
            }else if(conn->type == PEER_CONNECTION){
                val = peerConnHandler(conn, buffer, ret);
            }
            if(val == -1){
                break;
            }
        }
        //此处应判断定时器是否到期
    }

}


void Dispatcher::writeEventHandler(connection_t* conn)
{
    //std::cout << "写事件就绪" <<std::endl;
    auto logger = utils.logger.setBasicLogger("boss");
    int ret;
    if(conn == nullptr) return;
    responder* writer = conn->writer.get();
    if(writer == nullptr) return;
    Socket temp(writer->fd, CLIENT_SOCKET);
    {
        std::lock_guard<std::mutex>(writer->responder_lock); //获取responder_lock，防止线程竞争
        if(writer->bytes_left == -1 || writer->bytes_total == -1)
            return;
        while((ret = temp.Send(writer->current, writer->bytes_left)) > 0){
            writer->bytes_left -= ret;
            writer->current += ret;
            if(writer->bytes_left == 0)
                break;
        }
    }
    std::cout << "写事件处理完毕" <<std::endl;
    if(writer->bytes_left == 0)
    {
        //发送完毕，将写事件从epoll中移除
        conn->ev ^= EPOLLOUT; 
        epoll_event ev = utils.IOMultiplexer.setEpollEvent(conn->fd, conn->ev, (void*)conn);
        utils.IOMultiplexer.epollAddEvent(ev, conn->fd);
        conn->writer.reset(); //释放writer
    } 
    if(ret < 0){
        if(temp.getErrorno() != EAGAIN ){
            logger->error("send()失败，原因是：{}", MyLogger::getSysError().c_str());
            conn->ev ^= EPOLLOUT; 
            epoll_event ev = utils.IOMultiplexer.setEpollEvent(conn->fd, conn->ev, (void*)conn);
            utils.IOMultiplexer.epollAddEvent(ev, conn->fd);
            conn->writer.reset(); //释放writer
        }
    }
    
}


//用户连接就绪时的处理函数
int Dispatcher::localConnHandler(connection_t* conn,const char* buffer, int bytes)
{
    //处理协议
    if(conn->reader->parser(buffer, bytes) == false){
        //用户发送数据有误，向客户端发送反馈信息 
        json res;
        res["status"] = 400;
        res["msg"] = "错误的协议格式";
        string data = res.dump();
        echo(data, conn, utils);
        conn->reader->reset(); //用户数据出错时重置协议
        return -1;
    }else{
        if(conn->reader->isComplete()){
            std::cout << "协议解析完毕，socketfd: "<< conn->fd << " request-resource: "<<conn->reader->dataSet["request-resource"]<<" content-length: " <<conn->reader->dataSet["content-length"] << " data: " <<conn->reader->dataSet["data"]<<std::endl;
        }else{
            std::cout << "协议解析未完成，socketfd: " << conn->fd << std::endl;
        }
        if(!conn->reader->isComplete()) 
            return 0;
        else{
            std::map<string, string> dataSet = conn->reader->dataSet;//协议解析完成，获取用户数据
            conn->reader->reset();//重置协议状态以进行下一次协议解析
            biz* temp = biz::getInstance(utils);
            workerPool.enqueue(&biz::userReqHandler, temp, dataSet, conn);
            return 1;
        }
    }
}


//对端连接就绪时的处理函数
int Dispatcher::peerConnHandler(connection_t* conn, const char* buffer, int bytes)
{
    auto logger = utils.logger.setBasicLogger("Dispatcher");
    int start = 0;
    int res;
    //这里引入start的原因是在调试时候发现，如果同一个连接在短时间内多次发送数据，
    //epoll只会返回一次，这个时候读取的数据是多次请求的集合。因此我们必须考虑
    //多次请求集合在一起的情况下的协议解析
    while((res=conn->peerReader->parser(buffer, bytes, start)) > 0){
        string haha(buffer);
        if(conn->peerReader->isComplete()){
        std::cout << "对端连接协议解析完毕，socketfd: "<< conn->fd << " data: " << conn->peerReader->data <<std::endl;
        }else{
            std::cout << "协议解析未完成，socketfd: " << conn->fd << std::endl;
        }
        start = res;
        if(!conn->peerReader->isComplete()){
            return 0;
        }else{
            string data = conn->peerReader->data;//协议解析完成，获取用户数据
            conn->peerReader->reset();//重置协议状态以进行下一次协议解析
            biz* temp = biz::getInstance(utils);
            workerPool.enqueue(&biz::peerReqHandler, temp, data, conn);
        }
    }
    if(res < 0){
        logger->error("检测到解释器端错误的协议格式");
        conn->peerReader->reset();
        return -1;
    }else{
        if(conn->peerReader->isComplete()){
        std::cout << "对端连接协议解析完毕，socketfd: "<< conn->fd << " data: " << conn->peerReader->data <<std::endl;
        }else{
            std::cout << "协议解析未完成，socketfd: " << conn->fd << std::endl;
        }
        if(!conn->peerReader->isComplete()){
            return 0;
        }else{
            string data = conn->peerReader->data;//协议解析完成，获取用户数据
            conn->peerReader->reset();//重置协议状态以进行下一次协议解析
            biz* temp = biz::getInstance(utils);
            workerPool.enqueue(&biz::peerReqHandler, temp, data, conn);
            return 1;
        }
    }
}